package com.fanxuankai.boot.mqbroker.consumer;

import cn.hutool.core.net.NetUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fanxuankai.boot.mqbroker.consumer.*;
import com.fanxuankai.boot.mqbroker.consumer.autoconfigure.MqBrokerConsumerProperties;
import com.fanxuankai.boot.mqbroker.Msg;
import com.fanxuankai.boot.mqbroker.producer.SendStatus;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author fanxuankai
 */
public class MsgReceiveServiceImpl extends ServiceImpl<MsgReceiveMapper, MsgReceive> implements MsgReceiveService {
    @Resource
    private MqBrokerConsumerProperties properties;
    @Resource
    private MsgReceiveConsumer msgReceiveConsumer;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;
    @Resource
    private EventListenerRegistry eventListenerRegistry;

    @Override
    public List<MsgReceive> pullData() {
        if (eventListenerRegistry.getTopics().isEmpty()) {
            return Collections.emptyList();
        }
        return page(new Page<>(1, properties.getCompensation()
                .getBatchSize()), Wrappers.lambdaQuery(MsgReceive.class)
                .eq(Msg::getStatus, ReceiveStatus.WAIT.getCode())
                .in(Msg::getTopic, eventListenerRegistry.getTopics()).orderByAsc(MsgReceive::getId)
                .lt(MsgReceive::getRetry, properties.getMaxRetry())).getRecords();
    }

    @Override
    public boolean lock(Long id) {
        MsgReceive entity = new MsgReceive();
        entity.setStatus(ReceiveStatus.CONSUMING.getCode());
        entity.setLastModifiedDate(new Date());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        return update(entity, Wrappers.lambdaUpdate(MsgReceive.class).eq(Msg::getStatus, ReceiveStatus.WAIT.getCode())
                .eq(Msg::getId, id));
    }

    @Override
    public void consumeTimeout() {
        Date timeout = Date.from(LocalDateTime.now().minus(properties.getTimeout(), ChronoUnit.MILLIS)
                .atZone(ZoneId.systemDefault()).toInstant());
        int lastChance = properties.getMaxRetry();
        while (true) {
            Page<MsgReceive> page = page(new Page<>(1, 100),
                    Wrappers.lambdaQuery(MsgReceive.class)
                            .eq(Msg::getStatus, ReceiveStatus.CONSUMING.getCode())
                            .lt(Msg::getLastModifiedDate, timeout));
            if (page.getRecords().isEmpty()) {
                break;
            }
            MsgReceive entity = new MsgReceive();
            entity.setCause("消费超时");
            entity.setLastModifiedDate(new Date());
            for (MsgReceive msgReceive : page.getRecords()) {
                entity.setStatus(msgReceive.getRetry() < lastChance ? SendStatus.WAIT.getCode() :
                        SendStatus.FAILURE.getCode());
                update(entity, Wrappers.lambdaUpdate(MsgReceive.class)
                        .eq(Msg::getStatus, ReceiveStatus.CONSUMING.getCode())
                        .lt(msgReceive.getRetry() < lastChance, Msg::getRetry, lastChance)
                        .ge(msgReceive.getRetry() >= lastChance, Msg::getRetry, lastChance)
                        .lt(Msg::getLastModifiedDate, timeout));
            }
            if (!page.hasNext()) {
                break;
            }
        }
    }

    @Override
    public void success(MsgReceive msg) {
        MsgReceive entity = new MsgReceive();
        entity.setLastModifiedDate(new Date());
        entity.setStatus(ReceiveStatus.SUCCESS.getCode());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        update(entity, Wrappers.lambdaUpdate(MsgReceive.class).eq(Msg::getId, msg.getId())
                .eq(Msg::getStatus, ReceiveStatus.CONSUMING.getCode()));
    }

    @Override
    public void failure(MsgReceive msg) {
        MsgReceive entity = new MsgReceive();
        entity.setRetry(msg.getRetry());
        entity.setCause(msg.getCause());
        entity.setLastModifiedDate(new Date());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        if (msg.getRetry() < properties.getMaxRetry()) {
            entity.setStatus(ReceiveStatus.WAIT.getCode());
        } else {
            entity.setStatus(ReceiveStatus.FAILURE.getCode());
        }
        update(entity, Wrappers.lambdaUpdate(MsgReceive.class).eq(Msg::getId, msg.getId())
                .eq(Msg::getStatus, ReceiveStatus.CONSUMING.getCode()));
    }

    @Override
    public void updateRetry(MsgReceive msg) {
        MsgReceive entity = new MsgReceive();
        entity.setCause(msg.getCause());
        entity.setRetry(msg.getRetry());
        entity.setLastModifiedDate(new Date());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        update(entity, Wrappers.lambdaUpdate(MsgReceive.class).eq(Msg::getId, msg.getId()));
    }

    @Override
    public void consume(MsgReceive msg, boolean retry, boolean async) {
        if (async) {
            threadPoolExecutor.execute(() -> msgReceiveConsumer.consume(msg, retry));
        } else {
            msgReceiveConsumer.consume(msg, retry);
        }
    }
}
